package com.tkjt.pms.calc.core;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.tkjt.pms.calc.CalcTraceUpdater;
import com.tkjt.pms.calc.FormulaExecutor;
import com.tkjt.pms.calc.KpiCalculator;
import com.tkjt.pms.calc.common.IndexCalcConsts;
import com.tkjt.pms.calc.common.IndexCalcParallel;
import com.tkjt.pms.calc.common.IndexCalcSerial;
import com.tkjt.pms.calc.core.task.CalcTask;
import com.tkjt.pms.calc.utils.EsCalcUtils;
import com.tkjt.pms.common.IndexConsts;
import com.tkjt.pms.common.IndexParam;
import com.tkjt.pms.common.utils.EsJsonUtils;
import com.tkjt.pms.server.ServerService;
import com.tkjt.pms.server.common.ServerParam;
import com.tkjt.pms.utils.EsExceptionUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service(KpiCalculator.SERVICE_CALC_SERVER)
public class CalcServerService implements ServerService {

	@Autowired
	private FormulaExecutor formulaExecutor;
	@Autowired
	private CalcTraceUpdater calcTraceUpdater;

	private ExecutorService executorService = Executors.newFixedThreadPool(16);

	@Override
	public void doService(IndexParam indexParam) {
		log.info("server params: \r\n{}\r\n", EsJsonUtils.toJsonStr(indexParam));
		String indexYearMonth = (String) indexParam.getParam(IndexConsts.PARAM_INDEX_YEAR_MONTH);
		IndexCalcSerial indexCalcSerial = null;
		indexCalcSerial = (IndexCalcSerial) indexParam.getParam(IndexConsts.PARAM_INDEX_CALC_SERIAL);
		if (null == indexCalcSerial) {
			log.error("indexCalcParallelInfo==null, indexYearMonth={}", indexYearMonth);
			return;
		}
		String pgroupNo = EsCalcUtils.getPGroupCode(indexYearMonth);
		log.info("start calc parallel, pgroupNo={}, dependecies size={} ...", pgroupNo, indexCalcSerial.getParallelSize());
		indexParam.addParam(IndexCalcConsts.PARAM_CALC_PGROUP_CODE, pgroupNo);
		this.calcuParallel(indexCalcSerial, indexParam);
		log.info("end calc parallel, pgroupNo={}.", pgroupNo);
	}

	private void calcuParallel(IndexCalcSerial indexCalcSerial, IndexParam indexParam) {
		List<Future<Boolean>> taskFutures = new ArrayList<Future<Boolean>>();
		List<IndexCalcParallel> calcParallels = indexCalcSerial.getParallels();
		for (int index = 0; index < calcParallels.size(); index++) {
			ServerParam calcParallelIndexParam = new ServerParam();
			BeanUtils.copyProperties(indexParam,calcParallelIndexParam);
			Map<String,Object> param = new HashMap<>();
			param.putAll(indexParam.getParams());
			calcParallelIndexParam.addAllParam(param);
			CalcTask oneTask = new CalcTask(calcParallelIndexParam, this.formulaExecutor, calcParallels.get(index));
			Future<Boolean> future = executorService.submit(oneTask);
			taskFutures.add(future);
		}
		// 等待所有线程执行完成
		for (int index = 0; index < taskFutures.size(); index++) {
			IndexCalcParallel indexCalcParallel = calcParallels.get(index);
			String exceptionMessage = null;
			try {
				Boolean ret = taskFutures.get(index).get(10, TimeUnit.MINUTES);
				if(!ret) {
					log.error("calc task error, ret={}", ret);
				}
				continue;
			} catch (InterruptedException | ExecutionException | TimeoutException e) {
				log.error(e.getMessage(), e);
				exceptionMessage = EsExceptionUtils.formatException(e);
			}
			calcTraceUpdater.updateException(indexParam, indexCalcParallel, exceptionMessage);
		}
	}

}
